package com.yl.flink.processor;

import cn.hutool.core.util.ArrayUtil;
import com.yl.constant.Const;
import com.yl.constant.WarnSendType;
import com.yl.entity.PrimaryWarn;
import com.yl.entity.cdc.AuthUser;
import com.yl.util.FlinkUtil;
import com.yl.util.SUtil;
import com.yl.util.SendUtil;
import lombok.extern.slf4j.Slf4j;
import okhttp3.OkHttpClient;
import org.apache.flink.api.java.utils.ParameterTool;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.util.Collector;
import org.eclipse.paho.client.mqttv3.MqttClient;

import java.util.ArrayList;
import java.util.List;
import java.util.Map;

/**
 * @author wlf
 * @since 2022/8/19
 */
@Slf4j
public class WarnSendFunc extends ProcessFunction<PrimaryWarn, Object> {

    private transient ParameterTool params;
    private transient String projectKey;
    // mqtt
    private transient MqttClient mqttClient;
    private transient String mqttTopicPrefix;
    // ding
    private transient OkHttpClient httpClient;
    private transient String dingSignUrl;
    private transient String dingTitle;

    /**
     * 初始化资源配置
     */
    @Override
    public void open(Configuration parameters) throws Exception {
        params = FlinkUtil.getParameterTool(getRuntimeContext());
        mqttClient = FlinkUtil.getMqttClient(params);
        mqttTopicPrefix = params.get(Const.MQTT_TOPIC_WARN);
        httpClient = FlinkUtil.getDingClient(params);
        dingSignUrl = FlinkUtil.getDingSignUrl(params);
        dingTitle = params.get(Const.DING_BASE_PROJECT);
//        projectKey = params.get(Const.APP_NAME);
        projectKey = "cr11g_landslip";
    }

    /**
     * 推送告警信息
     *
     * @param primaryWarn 告警信息
     * @param context     上下文
     * @param out         输入器
     */
    @Override
    public void processElement(PrimaryWarn primaryWarn, Context context, Collector<Object> out) throws Exception {
        Map<Integer, List<AuthUser>> usersMap = getUsers(primaryWarn);
        if (params.getBoolean(Const.WARN_SEND_MQTT)) {
            // 主题格式：/项目编号/data/项目id/类型id/测点id
            String topic = SUtil.fmtMqttTopic(
                    mqttTopicPrefix,
                    primaryWarn.getMultiDataEntity().getProject().getId(),
                    primaryWarn.getMultiDataEntity().getTargetType().getId(),
                    primaryWarn.getMultiDataEntity().getTarget().getId());
            SendUtil.sendMqtt(mqttClient, topic, primaryWarn);
        }
        if (params.getBoolean(Const.WARN_SEND_DING)) {
            SendUtil.sendDing(httpClient, dingSignUrl, dingTitle, primaryWarn);
        }
        if (params.getBoolean(Const.WARN_SEND_VOICE)) {
            List<AuthUser> users = usersMap.get(WarnSendType.VOICE.getId());
            users.forEach(user -> SendUtil.sendVoice(user.getPhone(), primaryWarn));
        }
        if (params.getBoolean(Const.WARN_SEND_EMAIL)) {
            List<AuthUser> users = usersMap.get(WarnSendType.EMAIL.getId());
            users.forEach(user -> SendUtil.sendEmail(projectKey, user.getEmail(), primaryWarn));
        }
        if (params.getBoolean(Const.WARN_SEND_MESSAGE)) {
            List<AuthUser> users = usersMap.get(WarnSendType.MESSAGE.getId());
            users.forEach(user -> SendUtil.sendMsg(projectKey, user.getPhone(), primaryWarn));
        }
    }

    /**
     * 获取接收指定告警等级的所有用户
     * 是否接收告警（0否，1邮件，2短信，3语音），使用字段email和phone的值，多个空格分隔，使用中杠指定接收的告警等级（多个逗号分隔），
     * 若不指定则默认接收所有等级告警，例如：1 2-2,3 3-1   表示邮件接收所有等级告警，短信接收二级、三级告警，语音只接收一级告警
     *
     * @param primaryWarn 告警信息
     */
    private static Map<Integer, List<AuthUser>> getUsers(PrimaryWarn primaryWarn) {
        List<AuthUser> users = primaryWarn.getMultiDataEntity().getUsers();
        int level = primaryWarn.getWarn().getLevel();
        Map<Integer, List<AuthUser>> res = Map.of(
                WarnSendType.EMAIL.getId(), new ArrayList<>(),
                WarnSendType.MESSAGE.getId(), new ArrayList<>(),
                WarnSendType.VOICE.getId(), new ArrayList<>()
        );
        for (AuthUser user : users) {
            String receiveWarn = user.getReceive_warn();
            String[] allKinds = receiveWarn.split(Const.PTN_SPACE);
            for (String kindLel : allKinds) {
                String[] split = kindLel.split(Const.PTN_BAR_MID);
                int kind = Integer.parseInt(split[0]);
                if (split.length == 2) {
                    String[] levels = split[1].split(Const.PTN_COMMA);
                    if (ArrayUtil.contains(levels, String.valueOf(level))) {
                        res.get(kind).add(user);
                    }
                } else {
                    res.get(kind).add(user);
                }
            }
        }
        return res;
    }

    /**
     * 关闭配置资源
     */
    @Override
    public void close() throws Exception {
        if (mqttClient != null) {
            mqttClient.disconnect();
            mqttClient.close();
        }
    }

}
